package net.bwie.realtime.dwd.douyin.logs.job;

import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 交易订单支付明细数据清洗
 */

public class TradePayCleanJob {
    public static void main(String[] args) throws Exception {


        //1-执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 从kafka的ods订单支付明细数据消费
        DataStream<String> payStream = KafkaUtil.consumerKafka(env, "ods_trade_pay_detail");
//        payStream.print("pay");

        //3-数据清洗-json解析+字段校验+过滤
        SingleOutputStreamOperator<String> cleanStream = payStream.map(new MapFunction<String, String>() {
                    @Override
                    public String map(String s) throws Exception {
                        //json字符串解析
                        try{
                            return JSONObject.parseObject(s).toJSONString();
                        } catch (Exception e) {
                            return null;
                        }
                    }
                })
                //过滤空值
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        if(s != null){
                            return true;
                        }
                        return false;
                    }
                })
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        if(
                            //containsKey:判断key是否存在，存在再判断
                                jsonObject.containsKey("order_id") &&
                                        jsonObject.containsKey("user_id") &&
                                        jsonObject.containsKey("pay_id")
                        ){
                            return true;
                        }
                        return false;
                    }
                })
                // 数值字段校验：order_amount >= 0
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        if(
                            //金额不为null,>=0
                            jsonObject.getDouble("pay_amount") != null &&
                            jsonObject.getDouble("pay_amount") >= 0
                        ){
                            return true;
                        }
                        return false;
                    }
                })
                // 枚举校验
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        String pay_status = jsonObject.getString("pay_status");
                        return pay_status.equals("failed") || pay_status.equals("success");
                    }
                });

        //4. 打印清洗后的数据（后续可以写入dwdkafka）
        cleanStream.print("cleanPayDetail");

        KafkaUtil.producerKafka(cleanStream,"dwd_trade_pay");
        //5. 执行任务
        env.execute("TradePayCleanJob");
    }
}


